在 Day 21 到 Day 24 中,我們完成了情報從 Google Sheets 到 Obsidian Vault 以及 Notion 關聯資料庫的雙軌沉澱管線。
然而,當系統運作在 GitHub Actions 的無人值守環境時,一個嚴峻的工程挑戰浮現了:LLM 代理人管線的「思考黑盒子」問題。
當某篇論文被評估為 verified_score = 3 並遭到 ARCHIVED(封存)時,我們無從得知:
在分散式與 Agent 系統中,日誌(Logs)與追蹤(Traces)是維持系統可信任度與快速定位 Bad Cases 的核心支柱。今天我們將為情報雷達設計專屬的 Agent Trace 日誌架構,實現免外部第三方平台依賴的結構化思考鏈追蹤。
在 OpenTelemetry 的規範中,一個 Trace(呼叫鏈追蹤) 代表一筆情資從進站到出站的完整生命週期;而每個代理人的執行過程則是一個 Span(區間)。
我們在領域層為 IntelligenceItem 注入輕量級的追蹤上下文,避免引入龐大的 APM(Application Performance Monitoring)基礎設施:
Trace: entry_id="dfd9abe0"
│
├── Span 1: [Reviewer]
│ ├── Duration: 1,420ms | Tokens: 420
│ ├── Inputs: raw_content
│ ├── Decision: PASS (Score=5)
│ └── Reasoning: "確認包含完整實驗數據與多進程鎖定設計驗證..."
│
├── Span 2: [Domain Expert]
│ ├── Duration: 2,150ms | Tokens: 680
│ └── Extracted: metrics, limitations
│
├── Span 3: [Synthesizer]
│ ├── Duration: 1,890ms | Tokens: 530
│ └── Chain-of-Thought: "一階為並發控制 -> 二階推演出分散式狀態機在多代理協作之安全性..."
│
└── Span 4: [Editor]
├── Duration: 980ms | Tokens: 310
└── Output: editorial_summary, tags
在 src/core/models.py 中引入追蹤模型,使每個代理人能以不可變且結構化的格式回傳執行細節:
# src/core/models.py (節錄擴展)
from datetime import datetime
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field
class AgentSpan(BaseModel):
"""單一代理人執行節點的追蹤區間 (Span)"""
agent_name: str = Field(description="代理人名稱 (Reviewer, DomainExpert, Synthesizer, Editor)")
started_at: datetime = Field(default_factory=datetime.utcnow)
duration_ms: int = Field(description="執行耗時 (毫秒)")
prompt_tokens: int = Field(default=0, description="輸入 Token 數")
completion_tokens: int = Field(default=0, description="輸出 Token 數")
decision: Optional[str] = Field(default=None, description="核心決策 (例如 PASS / ARCHIVED)")
thought_process: Optional[str] = Field(default=None, description="思維鏈 (Chain-of-Thought) 摘要")
error: Optional[str] = Field(default=None, description="例外訊息 (若執行失敗)")
class ExecutionTrace(BaseModel):
"""單筆情資處理的完整追蹤鏈 (Trace)"""
trace_id: str
entry_id: str
total_duration_ms: int = 0
spans: List[AgentSpan] = Field(default_factory=list)
並在 IntelligenceItem 的 metadata 中保留 execution_trace 鍵名,維持現有 Pydantic 模型定義向下相容。
為了避免在每個代理人的業務邏輯中夾雜繁瑣的時間計算與日誌格式化代碼,我們採用 Python 裝飾器(Decorator) 模式,打造 @trace_agent_step:
# src/core/tracer.py
import functools
import logging
import time
from typing import Callable
from src.core.models import AgentSpan, IntelligenceItem
logger = logging.getLogger("agent.tracer")
def trace_agent_step(agent_name: str):
"""攔截代理人執行過程,自動記錄耗時、決策推導與思考鏈追蹤。"""
def decorator(func: Callable):
@functools.wraps(func)
def wrapper(self, item: IntelligenceItem, *args, **kwargs):
start_time = time.perf_counter()
error_msg = None
decision = None
thought_process = None
logger.info(f"[{item.entry_id}] 開始執行節點: {agent_name}")
try:
# 執行實際的代理人邏輯
result_item, should_continue = func(self, item, *args, **kwargs)
# 動態萃取關鍵代理人的決策與思考
if agent_name == "Reviewer":
decision = f"SCORE_{result_item.verified_score}" if result_item.verified_score else "ARCHIVED"
thought_process = result_item.metadata.get("review_reasoning")
elif agent_name == "Synthesizer":
decision = "SYNTHESIZED"
thought_process = result_item.cross_impact_notes
return result_item, should_continue
except Exception as exc:
error_msg = str(exc)
logger.error(f"[{item.entry_id}] 節點 {agent_name} 執行失敗: {exc}")
raise exc
finally:
duration_ms = int((time.perf_counter() - start_time) * 1000)
span = AgentSpan(
agent_name=agent_name,
duration_ms=duration_ms,
decision=decision,
thought_process=thought_process[:200] if thought_process else None,
error=error_msg
)
# 將 Span 寫入 IntelligenceItem 的追蹤歷史中
if "traces" not in item.metadata:
item.metadata["traces"] = []
item.metadata["traces"].append(span.model_dump(mode="json"))
logger.info(
f"[{item.entry_id}] 節點 {agent_name} 完成 "
f"| 耗時: {duration_ms}ms "
f"| 決策: {decision or 'OK'} "
f"| 錯誤: {error_msg is not None}"
)
return wrapper
return decorator
在 src/services/pipeline.py 中,代理人方法只需掛上裝飾器,即可自動完成指標採集:
# src/services/pipeline.py (節錄)
from src.core.tracer import trace_agent_step
class ReviewerNode:
@trace_agent_step("Reviewer")
def evaluate(self, item: IntelligenceItem) -> tuple[IntelligenceItem, bool]:
# 調用 LLM 評估邏輯
...
return item, True
class DomainExpertNode:
@trace_agent_step("DomainExpert")
def extract_metrics(self, item: IntelligenceItem) -> tuple[IntelligenceItem, bool]:
# 調用 LLM 指標萃取
...
return item, True
當管線處理完成一筆情報時,item.metadata["traces"] 會完整保留每個節點的耗時、決策依據與思考鏈。
追蹤資料收集完成後,我們提供兩種排查途徑:
在 pipeline.py 的批次結算尾端,將所有已處理情報的 Trace 聚合寫入 logs/traces_YYYY-MM-DD.jsonl:
import json
from pathlib import Path
def dump_traces(items: list[IntelligenceItem]):
log_dir = Path("logs/traces")
log_dir.mkdir(parents=True, exist_ok=True)
today_str = datetime.utcnow().strftime("%Y-%m-%d")
trace_file = log_dir / f"trace_{today_str}.jsonl"
with trace_file.open("a", encoding="utf-8") as f:
for item in items:
record = {
"entry_id": item.entry_id,
"source_title": item.source_title,
"status": item.status,
"verified_score": item.verified_score,
"traces": item.metadata.get("traces", [])
}
f.write(json.dumps(record, ensure_ascii=False) + "\n")
在 .github/workflows/patrol.yml 中,加入保存日誌的步驟。即使 GitHub Runner 執行後銷毀,我們仍可在 GitHub 網頁上下載該次巡邏的所有 Agent 思考脈絡:
- name: Upload Agent Execution Traces
if: always()
uses: actions/upload-artifact@v4
with:
name: agent-execution-traces
path: logs/traces/*.jsonl
retention-days: 7
現在,當我們在 Google Sheets 或 Discord 看到某筆情資被標記為 ARCHIVED 或 verified_score = 3 時,只要開啟該筆 JSONL 紀錄:
{
"entry_id": "4ad99832",
"source_title": "State Locking Verification Paper",
"status": "ARCHIVED",
"verified_score": 3,
"traces": [
{
"agent_name": "Reviewer",
"duration_ms": 1340,
"decision": "SCORE_3",
"thought_process": "論文缺乏多節點並發壓測數據,僅在本地 Mock 驗證,創新性與工程實用價值不足以跨越 4 分門檻。",
"error": null
}
]
}
代理人的扣分理由一目了然,徹底告別「不知道 LLM 為何這樣判」的黑盒子窘境。
今天我們建構了自主 Agent 系統的可觀測性基石:
AgentSpan 領域模型,規範了各代理人思考過程的標準欄位。@trace_agent_step 裝飾器,以無侵入式的方式自動收集執行耗時、決策推導與思考鏈摘要。然而,文字與 JSONL 檔案在快速排查時仍然偏向繁瑣。如果能在本機開啟一個輕量網頁,以時間軸與卡片形式直觀瀏覽每一次巡航的代理人決策與 Bad Cases,除錯效率將更上一層樓。
在 Day 26,我們將延伸今天的 Trace 成果:「本地可視化除錯:使用 Streamlit 快速打造 Agent Pipeline 戰情檢視器」,用不到 100 行代碼建構屬於自己的 Agent Inspector 儀表板!